package com.bdqn.spark.chapter05.accumulator

import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Spark04_Acc_WordCount {
  def main(args: Array[String]): Unit = {
    val sparConf = new SparkConf().setMaster("local").setAppName("Acc")
    val sc = new SparkContext(sparConf)

    val rdd = sc.makeRDD(List("hello", "spark", "hello", "hadoop"))

    // 创建一个累加器的对象
    val wcACC: MyAccumulator = new MyAccumulator()

    // 在 SparkContext 注册这个累加器：用在行动算子中
    sc.register(wcACC,"wcAcc")

    rdd.foreach(str =>{
      wcACC.add(str)
    })

    val value: mutable.Map[String, Long] = wcACC.value
    println(value)

    sc.stop()

  }
}

/**
 * 自己自定义一个累加器：word count  ("hello",2)
 *
 * AccumulatorV2父类有2个泛型：
 * IN ： 累加器的输入数据类型 ：String
 * OUT:  累加器的输出数据类型 : mutable.Map[String, Long]
 *
 * 实现累加器中 的  6 个方法
 */
class MyAccumulator extends AccumulatorV2[String, mutable.Map[String, Long]] {

  // 自定义一个 可变的Map对象
  private var wcMap: mutable.Map[String, Long] = mutable.Map[String, Long]();


  /**
   * 判断 累加器 中是否有数据
   *
   * @return
   */
  override def isZero: Boolean = wcMap.isEmpty

  /**
   * 复制一个新的累加器
   *
   * @return
   */
  override def copy(): AccumulatorV2[String, mutable.Map[String, Long]] = {
    val newACC: MyAccumulator = new MyAccumulator()
    newACC.wcMap = this.wcMap
    newACC
  }

  /**
   * 清空累加器中的数据
   */
  override def reset(): Unit = {
    wcMap.clear()
  }

  /**
   * 把一个字符串添加到 累加器 中
   *
   * @param v
   */
  override def add(v: String): Unit = {
    val newCnt: Long = wcMap.getOrElse(v, 0L) + 1
    wcMap.update(v, newCnt)
  }

  override def merge(other: AccumulatorV2[String, mutable.Map[String, Long]]): Unit = {
    val map1 = this.wcMap
    val map2 = other.value

    map2.foreach {
      case (word, cnt) => {
        val newCnt = map1.getOrElse(word, 0L) + cnt
        map1.update(word, newCnt)
      }
    }
  }

  // 返回 最终的结果
  override def value: mutable.Map[String, Long] = wcMap
}